package bean;

import com.alipay.sofa.jraft.util.Bits;
import com.alipay.sofa.jraft.util.BytesUtil;
import com.google.common.collect.Lists;
import io.vertx.core.buffer.Buffer;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang.ArrayUtils;
import thirdpart.bean.CmdPack;
import thirdpart.fetchserv.IFetchService;
import thirdpart.order.OrderCmd;
import thirdpart.order.OrderDirection;

import java.util.List;
import java.util.Map;
import java.util.TimerTask;

/**
 * 核心功能
 */
@Log4j2
@RequiredArgsConstructor
public class FetchTask extends TimerTask {

    @NonNull
    private SeqConfig config;

    @Override
    public void run() {
        //遍历网关 获取数据
        /**/
        if(!config.getNode().isLeader()){
            return;
        }
        Map<String, IFetchService> fetchServiceMap = config.getFetchServiceMap();
        if(MapUtils.isEmpty(fetchServiceMap)) {
            return;
        }

        List<OrderCmd> cmds = collectAllOrders(fetchServiceMap);
        if(CollectionUtils.isEmpty(cmds)) {
            return;
        }
        log.info(cmds);
        //todo 对数据进行排序
        cmds.sort((o1,o2)->{
            //第一种排序方法
            /*if(o1.timestamp>o2.timestamp){
                return 1;
            }else if (o1.timestamp<o2.timestamp){
                return -1;
            }else{
                if(o1.direction == OrderDirection.BUY){
                    if(o1.direction==o2.direction){
                        if(o1.price>o2.price) {
                            return -1;
                        }else if (o1.price<o2.price){
                            return 1;
                        }else{
                            if(o1.volume>o2.volume){
                                return -1;
                            }else if(o1.volume<o2.volume){
                                return 1;
                            }else{
                                return 0;
                            }
                        }
                    }else{
                        //方向不同 不影响成交结果 顺序不变
                        return 0;
                    }
                } else if (o1.direction==OrderDirection.SELL){
                    if(o1.direction==o2.direction){
                        if(o1.price < o2.price) {
                            return -1;
                        }else if (o1.price > o2.price){
                            return 1;
                        }else{
                            //量比较
                            if(o1.volume>o2.volume){
                                return -1;
                            }else if(o1.volume<o2.volume){
                                return 1;
                            }else{
                                return 0;
                            }
                        }
                    }else{
                        //方向不同 不影响成交结果 顺序不变
                        return 0;
                    }
                }else{
                    return 1;
                }
            }*/

            //第二种写法 时间优先 价格优先
            int res = compareTime(o1,o2);
            if(res!=0){
                return res;
            }
            res = comparePrice(o1,o2);
            if(res!=0){
                return res;
            }
            res = compareVolume(o1,o2);
            return res;
        });

        //存储到KVStore 发送到撮合核心
        try {
            //1 生成 packetno
            long packetno = getPacketNoFromStore();

            //2 入库
            CmdPack cmdPack = new CmdPack(packetno, cmds);
            byte[] serialize = config.getCodec().serialize(cmdPack);
            insertToKvStore(packetno,serialize);

            //3 更行packetNo+=1
            updatePacketNoInStore(packetno+1);

            //4 发送
            config.getMulticastSender().send(
                    Buffer.buffer(serialize),
                    config.getMulticastPort(),
                    config.getMulticastIp(),
                    asyncResult -> {
                        System.out.println("fetchTask Send succeeded? " + asyncResult.succeeded());
                    }
            );
        }catch (Exception e){
            log.error("encode cmd packet error",e);
        }
    }

    private void updatePacketNoInStore(long packetno) {
        final byte[] bytes = new byte[8];

        Bits.putLong(bytes,0,packetno);
        config.getNode().getRheaKVStore().put(PACKET_NO_KEY,bytes);
    }

    private void insertToKvStore(long packetno, byte[] serialize) {
        byte[] key = new byte[8];
        Bits.putLong(key,0,packetno);
        /*put方法时异步的  bPut方法是同步的*/
        config.getNode().getRheaKVStore().put(key,serialize);

    }

    private static final byte[] PACKET_NO_KEY= BytesUtil.writeUtf8("seq_pachet_no");

    /**
     * 获取PacketNo
     * @return
     */
    private long getPacketNoFromStore() {
        final byte[] bPacketNo = config.getNode().getRheaKVStore().bGet(PACKET_NO_KEY);
        long packetNo = 0;
        if(ArrayUtils.isNotEmpty(bPacketNo)){
            packetNo= Bits.getLong(bPacketNo,0);
        }
        return packetNo;
    }

    private int compareVolume(OrderCmd o1, OrderCmd o2) {
        if(o1.volume>o2.volume){
            return -1;
        }else if(o1.volume<o2.volume){
            return 1;
        }else{
            return 0;
        }
    }

    private int comparePrice(OrderCmd o1, OrderCmd o2) {
        if(o1.direction==o2.direction){
            if(o1.price>o2.price){
                return o1.direction==OrderDirection.BUY?-1:1;
            }else if(o1.price<o2.price){
                return o1.direction==OrderDirection.SELL?-1:1;
            }else{
                return 0;
            }
        }else{
            return 0;
        }
    }

    private int compareTime(OrderCmd o1, OrderCmd o2) {
        if(o1.timestamp>o2.timestamp){
            return 1;
        }else if (o1.timestamp<o2.timestamp){
            return -1;
        }else {
            return 0;
        }
    }

    private List<OrderCmd> collectAllOrders(Map<String, IFetchService> fetchServiceMap) {
        //性能不可控 不推荐  不方便调试
        /*List<OrderCmd> orderCmds = fetchServiceMap.values().stream()
                .map(IFetchService::fetchData)
                .filter(CollectionUtils::isEmpty)
                .flatMap(List::stream)
                .collect(Collectors.toList());*/

        List<OrderCmd> orderCmds = Lists.newArrayList();

        for (IFetchService fetchService : fetchServiceMap.values()) {
            List<OrderCmd> cmds = fetchService.fetchData();
            if(CollectionUtils.isNotEmpty(cmds)) {
                orderCmds.addAll(cmds);
            }
        }
        return orderCmds;
    }
}
